package kp.rpc.registry;

import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.vertx.core.impl.ConcurrentHashSet;
import kp.rpc.RpcApplication;
import kp.rpc.cache.CustomCache;
import kp.rpc.cache.factory.CustomCacheFactory;
import kp.rpc.config.RegistryConfig;
import kp.rpc.model.ServiceMetaInfo;
import kp.rpc.utils.JsonUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @author tzf
 * etcd 服务注册中心
 */
@Slf4j
public class EtcdRegistry implements Registry {
    private Client client;
    private KV kvClient;
    /**
     * @description: etcd根节点
     */

    private static final String ETCD_ROOT_PATH = "/rpc/";
    // 注册中心key缓存
    private static final Set<String> registerSet = new HashSet<>(10);
    //注册中心缓存
    private final CustomCache REGISTRY_SERVICE_CACHE = CustomCacheFactory.getCustomCache(RpcApplication.getRpcConfig().getCache());
    //监听中心
    private final Set<String> watchSet = new ConcurrentHashSet<>(10);

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        //创建lease租约客户端
        Lease leaseClient = this.client.getLeaseClient();
        //创建一个600s的租约
        long leaseId = leaseClient.grant(600).get().getID();
        //设置要存储的键值对
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(Objects.requireNonNull(JsonUtil.toJsonStr(serviceMetaInfo)), StandardCharsets.UTF_8);
        //将键值与租约关联起来,并设置过期时间
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, value, putOption).get();
        //将registerKey添加到set集合中
        registerSet.add(registerKey);
    }

    @SneakyThrows
    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String key = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
        registerSet.remove(key);
    }

    @Override
    public void init(RegistryConfig registryConfig) {
        this.client = Client.builder().endpoints(registryConfig.getRegisterAddress()).connectTimeout(Duration.ofMillis(registryConfig.getTimeout())).build();
        this.kvClient = client.getKVClient();
        //心跳机制
        this.healthCheck();
        //注册主动下线hook->在JVM关闭之前做销毁操作
        Runtime.getRuntime().addShutdownHook(new Thread(this::destroy));
    }

    @Override
    public void destroy() {
        log.info("当前节点与{}时下线", LocalDateTime.now());
        registerSet.forEach(key -> {
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
                log.info("{}下线", key);
            } catch (Exception e) {
                throw new RuntimeException(String.format("%s服务下线失败!", key));
            }
        });
        if (kvClient != null) {
            kvClient.close();
        }
        //最后再关闭client客户端
        if (client != null) {
            client.close();
        }
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 读取缓存
        List<ServiceMetaInfo> ServiceCacheList = REGISTRY_SERVICE_CACHE.readCache(serviceKey);
        if (!ServiceCacheList.isEmpty()) {
            log.warn("{}==>读取缓存", serviceKey);
            return ServiceCacheList;
        }
        //前缀搜索,结尾一定要加'/'
        String prefix = ETCD_ROOT_PATH + serviceKey + "/";
        GetOption getOption = GetOption.builder().isPrefix(true).build();
        try {
            //前缀查询
            List<KeyValue> kvList = kvClient.get(ByteSequence.from(prefix, StandardCharsets.UTF_8), getOption).get().getKvs();
            List<ServiceMetaInfo> serviceMetaInfoList = kvList.stream().map(keyValue -> {
                String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                //监听key
                watch(keyValue.getKey().toString(StandardCharsets.UTF_8));
                return JsonUtil.toBean(value, ServiceMetaInfo.class);
            }).collect(Collectors.toList());
            //写入缓存
            REGISTRY_SERVICE_CACHE.writeCache(serviceKey, serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException("获取服务列表失败!检查网络或者注册中心是否正确");
        }
    }

    /**
     * @description: 心跳检测机制
     * @author tzf
     * @date: 2024/6/5 14:16
     */

    @Override
    public void healthCheck() {
        CronUtil.schedule("*/30 * * * * *", (Task) () -> {
            //遍历set集合
            registerSet.forEach(key -> {
                try {
                    List<KeyValue> kvList = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get().getKvs();
                    if (kvList == null || kvList.isEmpty()) {
                        log.info("{}节点已下线", key);
                        registerSet.remove(key);
                    } else {
                        log.info("{}节点正常,开始续期", key);
                        kvList.forEach(kv -> {
                            String valueJSON = kv.getValue().toString(StandardCharsets.UTF_8);
                            ServiceMetaInfo reRegisterServiceMetaInfo = JsonUtil.toBean(valueJSON, ServiceMetaInfo.class);
                            //设置续期时间
                            Objects.requireNonNull(reRegisterServiceMetaInfo).setRenewalTime(LocalDateTime.now());
                            try {
                                register(reRegisterServiceMetaInfo);
                            } catch (Exception e) {
                                log.info("{}服务续期失败!", kv.getKey());
                            }
                        });
                    }
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(key + "服务续签失败!");
                }
            });
        });
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(String serviceNodeKey) {
        boolean watch = watchSet.add(serviceNodeKey);
        if (watch) {
            //创建watch客户端
            Watch watchClient = client.getWatchClient();
            //实时监听
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                response.getEvents().forEach(event -> {
                    //switch 对DELETE PUT 进行判断处理
                    switch (event.getEventType()) {
                        case DELETE -> {
                            //清除对应的缓存
                            REGISTRY_SERVICE_CACHE.clearCache(serviceNodeKey);
                        }
                        case PUT, UNRECOGNIZED -> {

                        }
                        default -> {
                        }
                    }
                });
            });
        }
    }
}
